#include "EventLoop.hpp"

EventLoop::EventLoop(Acceptor & acceptor)
: _epfd(createEpollFd())
, _evtList(1024)
, _isLooping(false)
, _acceptor(acceptor)
, _conns()
, _evtfd(createEventFd())
, _pendings()
, _mtx()
{   
    // 监听listenFd
    addFdToEpoll(_acceptor._sock.getFd());

    // 监听eventfd
    addFdToEpoll(_evtfd);
}

EventLoop::~EventLoop()
{
    close(_epfd);
    close(_evtfd);
}

void EventLoop::loop(){
    _isLooping = true;
    cout << "EventLoop is running..." << endl;
    while(_isLooping){
        waitEpollFd();
    }
}

void EventLoop::unloop(){
    _isLooping = false;
    cout << "EventLoop is unlooping..." << endl;
}

void EventLoop::setNewConnectionCallback(TcpConnectionCallback && cb){
    _onNewConnection = std::move(cb);
}

void EventLoop::setMessageCallback(TcpConnectionCallback && cb){
    _onMessage = std::move(cb);
}

void EventLoop::setCloseCallback(TcpConnectionCallback && cb){
    _onClose = std::move(cb);
}

int EventLoop::createEpollFd(){
    int epfd = ::epoll_create(1);
    if(epfd < 0){
        throw "epoll_create error";
    }
    return epfd;
}

void EventLoop::addFdToEpoll(int fd){
    struct epoll_event ev;
    ev.data.fd = fd;
    ev.events = EPOLLIN;

    int ret = ::epoll_ctl(_epfd, EPOLL_CTL_ADD, fd, &ev);
    if(ret < 0){
        throw "EPOLL_CTL_ADD error";
    }
}

void EventLoop::delFdFromEpoll(int fd){
    int ret = ::epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, NULL);
    if(ret < 0){
        throw "EPOLL_CTL_DEL error";
    }
}

void EventLoop::waitEpollFd(){
    int readyNum = 0;
    do{
        readyNum = ::epoll_wait(_epfd, &_evtList[0], _evtList.size(), 5000);  // 5s超时
    } while (readyNum == -1 && errno == EINTR); // 中断信号

    if(readyNum < 0){
        throw "epoll_wait error：" + std::string(strerror(errno));
    }
    else if(readyNum == 0){  // 监听超时返回
        if(_isLooping){
            cout << "epoll_wait timeout" << endl;
        }
    }
    else{
        if((size_t)readyNum == _evtList.size()){
            // 扩容
            _evtList.resize(_evtList.size() * 2);
        }

        for(int i=0; i<readyNum; ++i){
            int fd = _evtList[i].data.fd;
            int listenFd = _acceptor._sock.getFd();

            if(fd == listenFd){ // 有新的连接
                handleNewConnection();
            }
            else if(fd == _evtfd){  // 子线程的任务到达
                doPendingFunctors();
            }
            else{ // 有已连接客户的消息
                handleMessage(fd);
            }
        }
    }
}

void EventLoop::handleNewConnection(){
    // 取出新连接
    int connfd = _acceptor.accept();
    // 加入epoll监听
    addFdToEpoll(connfd); 
    // 创建连接对象，并为其设置回调函数
    TcpConnectionPtr conn(new TcpConnection(connfd, this));
    conn->setNewConnectionCallback(_onNewConnection);
    conn->setMessageCallback(_onMessage);
    conn->setCloseCallback(_onClose);
    // 将新建连接对象，加入已连接客户映射
    _conns[connfd] = conn;

    // 其他补充操作：由回调函数实现
    conn->handleNewConnectionCallback();
}  

void EventLoop::handleMessage(int fd){
    auto it = _conns.find(fd);  // 查找连接对象 
    if(it != _conns.end()){ 
        // 客户端主动关闭连接
        if(it->second->isClosed()){
            // 回调函数：补充对于客户端主动断开连接的处理
            it->second->handleCloseCallback();

            delFdFromEpoll(fd);  // 取消对该客户的监听
            _conns.erase(it);    // 从客户连接管理map中删除
            return;
        }

        // 正常处理客户消息
        it->second->handleMessageCallback();
    }
}

int EventLoop::createEventFd(){
    int evtfd = ::eventfd(0, 0);
    if(evtfd < 0){
        throw "evenfd create error: " + std::string(strerror(errno));
    }
    return evtfd;
}

void EventLoop::handleRead(){
    uint64_t num;
    size_t ret = read(_evtfd, &num, sizeof(uint64_t));
    if(ret != sizeof(uint64_t)){
        throw "handleRead: " + std::string(strerror(errno));
    }
} 

void EventLoop::wakeup(){
    uint64_t num = 1;
    size_t ret = write(_evtfd, &num, sizeof(uint64_t));
    if(ret != sizeof(uint64_t)){
        throw "wakeup: " + std::string(strerror(errno));
    }
} 

void EventLoop::doPendingFunctors(){
    vector<Functor> tempTasks;  // 临时容器
    {// 加锁
        lock_guard<mutex> lkgd(_mtx); 
        tempTasks.swap(_pendings); // 一次性取走_pendings中所有内容
        handleRead();
    }// 解锁
    
    for(auto & task : tempTasks){ // 单独处理所有send()任务
        #if DEBUG
        cout << "before sendTask" << endl;
        #endif

        task();
        
        #if DEBUG
        cout << "after sendTask" << endl;
        #endif
    }
} 

void EventLoop::runInLoop(Functor && cb){
    {
        lock_guard<mutex> lkgd(_mtx); // 互斥
        _pendings.push_back(std::move(cb));
    }
    wakeup();  // 这一步可以不互斥，多个子线程同时做唤醒，不会有影响
               // 因为主线程会一次取走所有pendings内容，不会因为少接收一次wakeup而遗漏内容
}